package com.movie.theatre.web.producer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.movie.theatre.web.exception.CinemaException;
import com.movie.theatre.web.exception.ErrorCode;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.Charsets;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * @Auther: liwu
 * @Date: 2019/12/24 17:46
 * @Description:
 */
@Slf4j
@Component
public class MqSendService {

    @Autowired
    private TransactionMQProducer mqProducer;


    /**
     * 发送普通消息
     * @param msg
     * @param topic
     * @param tags
     * @throws CinemaException
     */
    public void sendNormalMessage(Object msg, String topic, String tags) throws CinemaException {

        String uuid = UUID.randomUUID().toString();

        log.info("将要发送的普通MQ消息Topic为"+ topic + "发送的消息内容为"+ JSON.toJSONString(msg));

        try {
            sendNormal(JSONObject.toJSONString(msg).getBytes(Charsets.UTF_8), topic, tags, uuid);

        } catch (Exception ex) {
            log.error("发送MQ消息异常", ex);
            throw new CinemaException(ErrorCode.MQ_SEND_ERROR);
        }
    }


    private void sendNormal(byte[] data, String topic, String tags, String serialNo) throws Exception {

        Message message = new Message(topic, tags, serialNo, data);
        SendResult sendResult = mqProducer.send(message);
        log.info("send rocketMQ message success; {}", sendResult);
        if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
            log.error("发送MQ消息异常[{}]", sendResult.getSendStatus());
            throw new CinemaException(ErrorCode.MQ_SEND_ERROR);
        }
    }
}
